/*
 * Copyright 2009 Armando Blancas
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package taskgraph.tasks.data;

import java.io.EOFException;
import java.io.IOException;
import java.io.InterruptedIOException;

import taskgraph.ports.BooleanInputPort;
import taskgraph.ports.ByteInputPort;
import taskgraph.ports.DoubleInputPort;
import taskgraph.ports.FloatInputPort;
import taskgraph.ports.InputPort;
import taskgraph.ports.IntInputPort;
import taskgraph.ports.LongInputPort;
import taskgraph.ports.ShortInputPort;
import taskgraph.tasks.PrimitiveTask;
import taskgraph.tasks.data.NativeBuffers.BooleanBuffer;
import taskgraph.tasks.data.NativeBuffers.ByteBuffer;
import taskgraph.tasks.data.NativeBuffers.DoubleBuffer;
import taskgraph.tasks.data.NativeBuffers.FloatBuffer;
import taskgraph.tasks.data.NativeBuffers.IntBuffer;
import taskgraph.tasks.data.NativeBuffers.LongBuffer;
import taskgraph.tasks.data.NativeBuffers.ShortBuffer;


/**
 * Writes a native array from elements read from the input port.
 * 
 * <p>The output array may be of any native type. The input port may be
 * the appropriate port for that native type, or an InputPort<Object>
 * where each reference is a wrapper for a native type.
 * 
 * <p>The array is available after the graph's execution through the
 * appropriate getter method. Clients may pass an already allocated
 * array to avoid a new allocation, but should take care in calling
 * getter method for the correct array reference, since there
 * is no guarantee that the passed array will have enough capacity
 * (in which case a new one is allocated). If the array has enough
 * capacity, that same reference is returned.
 * 
 * @author Armando Blancas
 */
public class WriteNativeArray extends PrimitiveTask {

	/*
	 * References to native arrays.
	 */
	private boolean[]booleanArray = new boolean[0];
	private byte[] byteArray      = new byte[0];
	private short[] shortArray    = new short[0];
	private int[] intArray        = new int[0];
	private long[] longArray      = new long[0];
	private float[] floatArray    = new float[0];
	private double[] doubleArray  = new double[0];
	
	/*
	 * References to type-specific output ports.
	 */
	private BooleanInputPort booleanInput;
	private ByteInputPort byteInput;
	private ShortInputPort shortInput;
	private IntInputPort intInput;
	private LongInputPort longInput;
	private FloatInputPort floatInput;
	private DoubleInputPort doubleInput;
	
	/*
	 * Generic output port for wrapper objects.
	 */
	private InputPort<Object> input;
	
	/**
	 * This constructor allows creating instances as beans.
	 */
    public WriteNativeArray() {
    	// do nothing
    }
    
    /**
     * WriteNativeArray constructor.
     * 
     * @param booleanInput An input port for transporting booleans.
     * @param booleanArray A {@code boolean} array to store data.
     */
	public WriteNativeArray(BooleanInputPort booleanInput, 
			                boolean[] booleanArray) {
		setBooleanInput(booleanInput);
		setBooleanArray(booleanArray);
    }

    /**
     * WriteNativeArray constructor.
     * 
     * @param byteInput An input port for transporting bytes.
     * @param byteArray A {@code byte} array to store data.
     */
	public WriteNativeArray(ByteInputPort byteInput, 
			                byte[] byteArray) {
		setByteInput(byteInput);
		setByteArray(byteArray);
    }

    /**
     * WriteNativeArray constructor.
     * 
     * @param doubleInput An input port for transporting doubles.
     * @param doubleArray A {@code double} array to store data.
     */
	public WriteNativeArray(DoubleInputPort doubleInput, 
			                double[] doubleArray) {
		setDoubleInput(doubleInput);
		setDoubleArray(doubleArray);
    }

    /**
     * WriteNativeArray constructor.
     * 
     * @param floatInput An input port for transporting floats.
     * @param floatArray A {@code float} array to store data.
     */
	public WriteNativeArray(FloatInputPort floatInput, 
			                float[] floatArray) {
		setFloatInput(floatInput);
		setFloatArray(floatArray);
    }

    /**
     * WriteNativeArray constructor.
     * 
     * @param input An input port for transporting wrapper objects.
     */
	public WriteNativeArray(InputPort<Object> input) {
		setInput(input);
    }

    /**
     * WriteNativeArray constructor.
     * 
     * @param intInput An input port for transporting integers.
     * @param intArray An {@code int} array to store data.
     */
	public WriteNativeArray(IntInputPort intInput, 
			                int[] intArray) {
		setIntInput(intInput);
		setIntArray(intArray);
    }

    /**
     * WriteNativeArray constructor.
     * 
     * @param longInput An input port for transporting long integers.
     * @param longArray A {@code long} array to store data.
     */
	public WriteNativeArray(LongInputPort longInput, 
			                long[] longArray) {
		setLongInput(longInput);
		setLongArray(longArray);
    }

    /**
     * WriteNativeArray constructor.
     * 
     * @param shortInput An input port for transporting short integers.
     * @param shortArray A {@code short} array to store data.
     */
	public WriteNativeArray(ShortInputPort shortInput, 
			                short[] shortArray) {
		setShortInput(shortInput);
		setShortArray(shortArray);
    }

	private void close() {
		if (booleanInput != null)
			booleanInput.close();
		if (byteInput != null)
			byteInput.close();
		if (shortInput != null)
			shortInput.close();
		if (intInput != null)
			intInput.close();
		if (longInput != null)
			longInput.close();
		if (floatInput != null)
			floatInput.close();
		if (doubleInput != null)
			doubleInput.close();
		if (input != null)
			input.close();
	}

	/**
	 * Gets the reference to the given boolean array.
	 * 
	 * @return The array reference.
	 */
	public boolean[] getBooleanArray() {
		return booleanArray;
	}

	/**
	 * Gets the reference to the given boolean port.
     *
	 * @return The boolean input port.
	 */
	public BooleanInputPort getBooleanInput() {
		return booleanInput;
	}

	/**
	 * Gets the reference to the given byte array.
	 * 
	 * @return The array reference.
	 */
	public byte[] getByteArray() {
		return byteArray;
	}

	/**
	 * Gets the reference to the given byte port.
	 * 
	 * @return The byte input port.
	 */
	public ByteInputPort getByteInput() {
		return byteInput;
	}

	/**
	 * Gets the reference to the given double array.
	 * 
	 * @return The array reference.
	 */
	public double[] getDoubleArray() {
		return doubleArray;
	}

	/**
	 * Gets the reference to the given double port.
	 * 
	 * @return The double input port.
	 */
	public DoubleInputPort getDoubleInput() {
		return doubleInput;
	}

	/**
	 * Gets the reference to the given float array.
	 * 
	 * @return The array reference.
	 */
	public float[] getFloatArray() {
		return floatArray;
	}

	/**
	 * Gets the reference to the given float port.
	 * 
	 * @return The float input port.
	 */
	public FloatInputPort getFloatInput() {
		return floatInput;
	}

	/**
     * Gets the input port, used for wrapper objects.
     * 
	 * @return The input port.
	 */
	public InputPort<Object> getInput() {
		return input;
	}

	/**
	 * Gets the reference to the given integer array.
	 * 
	 * @return The array reference.
	 */
	public int[] getIntArray() {
		return intArray;
	}

	/**
	 * Gets the reference to the given integer port.
	 * 
	 * @return The integer input port.
	 */
	public IntInputPort getIntInput() {
		return intInput;
	}

	/**
	 * Gets the reference to the given long integer array.
	 * 
	 * @return The array reference.
	 */
	public long[] getLongArray() {
		return longArray;
	}

	/**
	 * Gets the reference to the given long integer port.
	 * 
	 * @return The long input port.
	 */
	public LongInputPort getLongInput() {
		return longInput;
	}

	/**
	 * Gets the reference to the given short integer array.
	 * 
	 * @return The array port.
	 */
	public short[] getShortArray() {
		return shortArray;
	}

	/**
	 * Gets the reference to the given short integer array.
	 * 
	 * @return the shortInput
	 */
	public ShortInputPort getShortInput() {
		return shortInput;
	}

	private void loopBoolean() 
	throws InterruptedIOException, IOException {
		BooleanBuffer buffer = new BooleanBuffer();
		while (!Thread.currentThread().isInterrupted()) {
			try { 
				buffer.add(booleanInput.read()); 
			} catch (EOFException e) { 
				break; 
			}
		}
		booleanArray = buffer.toArray(booleanArray);
	}
	
	private void loopByte() 
	throws InterruptedIOException, IOException {
		ByteBuffer buffer = new ByteBuffer();
		while (!Thread.currentThread().isInterrupted()) {
			try {
				buffer.add(byteInput.read());
			} catch (EOFException e) {
				break;
			}
		}
		byteArray = buffer.toArray(byteArray);
	}
	
	private void loopDouble() 
	throws InterruptedIOException, IOException {
		DoubleBuffer buffer = new DoubleBuffer();
		while (!Thread.currentThread().isInterrupted()) {
			try {
				buffer.add(doubleInput.read());
			} catch (EOFException e) {
				break;
			}
		}
		doubleArray = buffer.toArray(doubleArray);
	}
	
	private void loopFloat() 
	throws InterruptedIOException, IOException {
		FloatBuffer buffer = new FloatBuffer();
		while (!Thread.currentThread().isInterrupted()) {
			try {
				buffer.add(floatInput.read());
			} catch (EOFException e) {
				break;
			}
		}
		floatArray = buffer.toArray(floatArray);
	}
	
	private void loopInput() 
	throws InterruptedIOException, IOException {
		Object obj = input.read();
		Class<?> clazz = obj.getClass();
		if (clazz == Boolean.class) {
			loopWrappedBoolean((Boolean) obj);
		} else if (clazz == Byte.class) {
			loopWrappedByte((Byte) obj);
		} else if (clazz == Short.class) {
			loopWrappedShort((Short) obj);
		} else if (clazz == Integer.class) {
			loopWrappedInt((Integer) obj);
		} else if (clazz == Long.class) {
			loopWrappedLong((Long) obj);
		} else if (clazz == Float.class) {
			loopWrappedFloat((Float) obj);
		} else if (clazz == Double.class) {
			loopWrappedDouble((Double) obj);
		}
	}
	
	private void loopInt() 
	throws InterruptedIOException, IOException {
		IntBuffer buffer = new IntBuffer();
		while (!Thread.currentThread().isInterrupted()) {
			try {
				buffer.add(intInput.read());
			} catch (EOFException e) {
				break;
			}
		}
		intArray = buffer.toArray(intArray);
	}
	
	private void loopLong() 
	throws InterruptedIOException, IOException {
		LongBuffer buffer = new LongBuffer();
		while (!Thread.currentThread().isInterrupted()) {
			try {
				buffer.add(longInput.read());
			} catch (EOFException e) {
				break;
			}
		}
		longArray = buffer.toArray(longArray);
	}
	
	private void loopShort() 
	throws InterruptedIOException, IOException {
		ShortBuffer buffer = new ShortBuffer();
		while (!Thread.currentThread().isInterrupted()) {
			try {
				buffer.add(shortInput.read());
			} catch (EOFException e) {
				break;
			}
		}
		shortArray = buffer.toArray(shortArray);
	}
	
	private void loopWrappedBoolean(boolean value) 
	throws InterruptedIOException, IOException {
		BooleanBuffer buffer = new BooleanBuffer();
		while (!Thread.currentThread().isInterrupted()) {
			try {
				buffer.add(value);
				value = (Boolean) input.read();
			} catch (EOFException e) { 
				break; 
			}
		}
		booleanArray = buffer.toArray(booleanArray);
	}

private void loopWrappedByte(byte value) 
	throws InterruptedIOException, IOException {
		ByteBuffer buffer = new ByteBuffer();
		while (!Thread.currentThread().isInterrupted()) {
			try {
				buffer.add(value);
				value = (Byte) input.read();
			} catch (EOFException e) { 
				break; 
			}
		}
		byteArray = buffer.toArray(byteArray);
	}
	
	private void loopWrappedDouble(double value) 
	throws InterruptedIOException, IOException {
		DoubleBuffer buffer = new DoubleBuffer();
		while (!Thread.currentThread().isInterrupted()) {
			try {
				buffer.add(value);
				value = (Double) input.read();
			} catch (EOFException e) { 
				break; 
			}
		}
		doubleArray = buffer.toArray(doubleArray);
	}
	
	private void loopWrappedFloat(float value) 
	throws InterruptedIOException, IOException {
		FloatBuffer buffer = new FloatBuffer();
		while (!Thread.currentThread().isInterrupted()) {
			try {
				buffer.add(value);
				value = (Float) input.read();
			} catch (EOFException e) { 
				break; 
			}
		}
		floatArray = buffer.toArray(floatArray);
	}
	
	private void loopWrappedInt(int value) 
	throws InterruptedIOException, IOException {
		IntBuffer buffer = new IntBuffer();
		while (!Thread.currentThread().isInterrupted()) {
			try {
				buffer.add(value);
				value = (Integer) input.read();
			} catch (EOFException e) { 
				break; 
			}
		}
		intArray = buffer.toArray(intArray);
	}
	
	private void loopWrappedLong(long value) 
	throws InterruptedIOException, IOException {
		LongBuffer buffer = new LongBuffer();
		while (!Thread.currentThread().isInterrupted()) {
			try {
				buffer.add(value);
				value = (Long) input.read();
			} catch (EOFException e) { 
				break; 
			}
		}
		longArray = buffer.toArray(longArray);
	}
	
	private void loopWrappedShort(short value) 
	throws InterruptedIOException, IOException {
		ShortBuffer buffer = new ShortBuffer();
		while (!Thread.currentThread().isInterrupted()) {
			try {
				buffer.add(value);
				value = (Short) input.read();
			} catch (EOFException e) { 
				break; 
			}
		}
		shortArray = buffer.toArray(shortArray);
	}
	
	/**
     * Performs the work of this task.
     *
     * Dispatch the loop processing method according to the specified port.
     */
	@Override
    public void run() {
    	try {
    		if (booleanInput != null) {
    			loopBoolean();
    		} else if (byteInput != null) {
    			loopByte();
    		} else if (shortInput != null) {
    			loopShort();
    		} else if (intInput != null) {
    			loopInt();
    		} else if (longInput != null) {
    			loopLong();
    		} else if (floatInput != null) {
    			loopFloat();
    		} else if (doubleInput != null) {
    			loopDouble();
    		} else if (input != null) {
    			loopInput();
    		} else {
                log.error("WriteNativeArray#run(): array not set");    			
    		}
        } catch (InterruptedIOException e) {
        	// interruption request
        } catch (ArrayIndexOutOfBoundsException e) {
        	// we're done reading the array
        } catch (IOException e) {
        	// pipe synch error
            log.error("WriteNativeArray#run()", e);
		} finally {
			close();
        }
    }

	/**
	 * Sets a boolean array.
	 * 
	 * @param booleanArray The boolean array to set.
	 */
	public void setBooleanArray(final boolean[] booleanArray) {
    	if (booleanArray == null)
    		throw new IllegalArgumentException("booleanArray == null");
		this.booleanArray = booleanArray;
	}

	/**
	 * Sets a boolean input port.
	 * 
	 * @param booleanInput The boolean input port to set.
	 */
	public void setBooleanInput(final BooleanInputPort booleanInput) {
    	if (booleanInput == null)
    		throw new IllegalArgumentException("booleanInput == null");
		this.booleanInput = booleanInput;
	}

	/**
	 * Sets a byte array.
	 * 
	 * @param byteArray The byte array to set.
	 */
	public void setByteArray(final byte[] byteArray) {
    	if (byteArray == null)
    		throw new IllegalArgumentException("byteArray == null");
		this.byteArray = byteArray;
	}

	/**
	 * Sets a byte input port.
	 * 
	 * @param byteInput The byte input port to set.
	 */
	public void setByteInput(ByteInputPort byteInput) {
    	if (byteInput == null)
    		throw new IllegalArgumentException("byteInput == null");
		this.byteInput = byteInput;
	}

	/**
	 * Sets a double array.
	 * 
	 * @param doubleArray The double array to set.
	 */
	public void setDoubleArray(double[] doubleArray) {
    	if (doubleArray == null)
    		throw new IllegalArgumentException("doubleArray == null");
		this.doubleArray = doubleArray;
	}

	/**
	 * Sets a double input port.
	 * 
	 * @param doubleInput The double input port to set.
	 */
	public void setDoubleInput(DoubleInputPort doubleInput) {
    	if (doubleInput == null)
    		throw new IllegalArgumentException("doubleInput == null");
		this.doubleInput = doubleInput;
	}

	/**
	 * Sets a float array.
	 * 
	 * @param floatArray The float array to set.
	 */
	public void setFloatArray(float[] floatArray) {
    	if (floatArray == null)
    		throw new IllegalArgumentException("floatArray == null");
		this.floatArray = floatArray;
	}

	/**
	 * Sets a float input port.
	 * 
	 * @param floatInput The float input port to set.
	 */
	public void setFloatInput(FloatInputPort floatInput) {
    	if (floatInput == null)
    		throw new IllegalArgumentException("floatInput == null");
		this.floatInput = floatInput;
	}

	/**
	 * Sets the input port, used for wrapper objects.
	 * 
	 * @param input The input port.
	 */
    public void setInput(final InputPort<Object> input) {
    	if (input == null)
    		throw new IllegalArgumentException("input == null");
 		this.input = input;
	}

	/**
	 * Sets an integer array.
	 * 
	 * @param intArray The int array to set.
	 */
	public void setIntArray(int[] intArray) {
    	if (intArray == null)
    		throw new IllegalArgumentException("intArray == null");
		this.intArray = intArray;
	}

	/**
	 * Sets an integer input port.
	 * 
	 * @param intInput The int input port to set.
	 */
	public void setIntInput(IntInputPort intInput) {
    	if (intInput == null)
    		throw new IllegalArgumentException("intInput == null");
		this.intInput = intInput;
	}


    /**
	 * Sets a long integer array.
	 * 
	 * @param longArray The long array to set.
	 */
	public void setLongArray(long[] longArray) {
    	if (longArray == null)
    		throw new IllegalArgumentException("longArray == null");
		this.longArray = longArray;
	}

	/**
     * Sets a long integer input port.
     * 
	 * @param longInput The long input port to set.
	 */
	public void setLongInput(LongInputPort longInput) {
    	if (longInput == null)
    		throw new IllegalArgumentException("longInput == null");
		this.longInput = longInput;
	}

	/**
	 * Sets a short array.
	 * 
	 * @param shortArray The short array to set.
	 */
	public void setShortArray(short[] shortArray) {
    	if (shortArray == null)
    		throw new IllegalArgumentException("shortArray == null");
		this.shortArray = shortArray;
	}

	/**
	 * Sets a short input port.
	 * 
	 * @param shortInput The short input port to set.
	 */
	public void setShortInput(ShortInputPort shortInput) {
    	if (shortInput == null)
    		throw new IllegalArgumentException("shortInput == null");
		this.shortInput = shortInput;
	}

}
